package com.wenx.kafkacamel;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.properties.PropertiesComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.SimpleRegistry;
import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * camel get message from kafka
 */
public final class MessageConsumerClient {

    private static final Logger LOG = LoggerFactory.getLogger(MessageConsumerClient.class);

    private MessageConsumerClient() {
    }

    public static void main(String[] args) throws Exception {

        LOG.info("About to run Kafka-camel integration...");
       /* KafkaIdempotentRepository kafkaIdempotentRepository =
                new KafkaIdempotentRepository("test", "10.10.5.212:9092");

        SimpleRegistry registry = new SimpleRegistry();
        registry.put("insertDbIdemRepo", kafkaIdempotentRepository); // must be registered in the registry, to enable access to the CamelContext*/

        CamelContext camelContext = new DefaultCamelContext();

        // Add route to send messages to Kafka

        camelContext.addRoutes(new RouteBuilder() {
            public void configure() {
                PropertiesComponent pc = getContext().getComponent("properties", PropertiesComponent.class);
                pc.setLocation("classpath:application-kafka.properties");

                log.info("About to start route: Kafka Server -> Log ");

               /* KafkaComponent kafka = new KafkaComponent();
                kafka.setBrokers("{{kafka.host}}:{{kafka.port}}");

                camelContext.addComponent("kafka", kafka);*/

                from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}"
                        + "&maxPollRecords={{consumer.maxPollRecords}}"
                        + "&consumersCount={{consumer.consumersCount}}"
                        + "&seekTo={{consumer.seekTo}}"
                        + "&groupId={{consumer.group}}")
                        .routeId("FromKafka")
                        .process(new Processor(){
                            public void process(Exchange exchange) throws Exception {
                                String aa = exchange.getIn().getBody().toString();
                                //System.out.println(aa);
                            }
                        })
                        .to("stream:out")
                        //.to("file:d:/logs/test.txt")
                        .log("${body}");

                /*from("kafka:10.10.5.212:9092?topic=test&groupId=testing&autoOffsetReset=earliest&consumersCount=1")
                        .process(new Processor() {
                            public void process(Exchange exchange)
                                    throws Exception {
                                String messageKey = "";
                                if (exchange.getIn() != null) {
                                    Message message = exchange.getIn();
                                    Integer partitionId = (Integer) message
                                            .getHeader(KafkaConstants.PARTITION);
                                    String topicName = (String) message
                                            .getHeader(KafkaConstants.TOPIC);
                                    if (message.getHeader(KafkaConstants.KEY) != null)
                                        messageKey = (String) message
                                                .getHeader(KafkaConstants.KEY);
                                    Object data = message.getBody();

                                    System.out.println("topicName :: "
                                            + topicName + " partitionId :: "
                                            + partitionId + " messageKey :: "
                                            + messageKey + " message :: "
                                            + data + "\n");
                                }
                            }
                        }).to("log:input");*/

                /*from("direct:start").process(new Processor() {

                    public void process(Exchange exchange) throws Exception {
                        exchange.getIn().setBody("Test Message from Camel Kafka Component Final",String.class);
                        exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0);
                        exchange.getIn().setHeader(KafkaConstants.KEY, "1");
                    }
                }).to("kafka:10.10.5.212:9092?topic=test");*/
            }
        });
        camelContext.start();

        // let it run for 5 minutes before shutting down
        Thread.sleep(5 * 60 * 1000);

        camelContext.stop();
    }

}